/**
 * Copyright (c) 2018, Mr.Wang (recallcode@aliyun.com) All rights reserved.
 */

package cn.recallcode.iot.mqtt.server.broker.protocol.impl;

import cn.recallcode.iot.mqtt.server.broker.protocol.AbstractMessageProcessor;
import cn.recallcode.iot.mqtt.server.common.message.DupPubRelMessageStore;
import cn.recallcode.iot.mqtt.server.common.message.IDupPubRelMessageStoreService;
import cn.recallcode.iot.mqtt.server.common.message.IDupPublishMessageStoreService;
import io.netty.channel.Channel;
import io.netty.handler.codec.mqtt.MqttFixedHeader;
import io.netty.handler.codec.mqtt.MqttMessage;
import io.netty.handler.codec.mqtt.MqttMessageFactory;
import io.netty.handler.codec.mqtt.MqttMessageIdVariableHeader;
import io.netty.handler.codec.mqtt.MqttMessageType;
import io.netty.handler.codec.mqtt.MqttQoS;
import io.netty.util.AttributeKey;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * PUBREC连接处理
 */
@Slf4j
@Data
@NoArgsConstructor
@AllArgsConstructor
public class PubRec extends AbstractMessageProcessor {
	private IDupPublishMessageStoreService dupPublishMessageStoreService;

	private IDupPubRelMessageStoreService dupPubRelMessageStoreService;

	@Override
	public void process(Channel channel, MqttMessage mqttMsg) {
		int messageId = ((MqttMessageIdVariableHeader) mqttMsg.variableHeader()).messageId();
		String clientId = (String) channel.attr(AttributeKey.valueOf("clientId")).get();
		MqttMessage pubRelMessage = MqttMessageFactory.newMessage(
			new MqttFixedHeader(MqttMessageType.PUBREL, false, MqttQoS.AT_MOST_ONCE, false, 0),
			MqttMessageIdVariableHeader.from(messageId), null);
		log.debug("PUBREC - clientId: {}, messageId: {}", clientId, messageId);
		dupPublishMessageStoreService.remove(clientId, messageId);
		DupPubRelMessageStore dupPubRelMessageStore = new DupPubRelMessageStore()
				.setClientId(clientId)
				.setMessageId(messageId);
		dupPubRelMessageStoreService.put(clientId, dupPubRelMessageStore);
		channel.writeAndFlush(pubRelMessage);
	}
}
